<?php
declare(strict_types=1);
namespace Service\ConsumerService;
use Service\BaseService;
use App\Model\PssLogModel\MessageSkuChangeLog;
use Hyperf\Utils\WaitGroup;
use App\Model\PssModel\Business;
use Service\Code\Normal;
use Service\HttpServer\Http;
use App\Amqp\Producer\PushBbcMessageProducer;
use Hyperf\Amqp\Producer;
use App\Amqp\Producer\PushThirdPartyBbcMessageProducer;
use Service\AppService\SystemCacheConfigService;
use App\Amqp\Producer\SkuUserChangeLogMessageProducer;

/**
 * 推送下游渠道商品消息变化中心
 * @author chenshubo
 */
class PushUserGoodsMessageService extends BaseService
{
    public $pageSize = 50;
    public $timeout  = 30;
    const SLEEP_TIMES_LIMIT = 3;
    
    /**
     * 获取需要直接通过API进行推送的商品数据
     * @param int $limit
     */
    public function queryMessageChangeLogLists(int $limit)
    {
        return MessageSkuChangeLog::instance()
        ->select([
            'id',
            'message_type',
            'bid',
            'push_url',
            'contents'
        ])
        ->orderBy('id','asc')
        ->limit($limit)
        ->get()
        ->toArray();
    }
    
    //SBBC 自营系统调用
    const default_self_bbc_api = 'http://127.0.0.1:9508/Api/Goods/queryIsChildYunCangGoodsLog';
    
    /**
     * 查询BBC渠道商用户
     */
    public function queryBbcGoodsMessage()
    {
        while ( true ) 
        {
            $userList = Business::instance()
            ->select([
                'id',
                'channel_id',
                'appid'
            ])
            ->where('is_channel',1)
            ->where('status',0)
            ->where('channel_id','>',0)
            ->get()
            ->toArray();
            
            if( empty($userList) ) {
                sleep(3);
                continue;
            }
            
            $wg = new WaitGroup();
            $wg->add(count($userList));
            $wg_result = [];
            foreach ($userList as $user)
            {
                co(function() use ($wg,$user,& $wg_result)
                {
                    //开始拉取供应商接口商品增量同步消息
                    $response = Http::postJsonCurl(self::default_self_bbc_api,['appid'=>$user['appid']]);
                    $result   = !empty($response) ? json_decode($response,true) : [];
                    
                    //取出数据之后，扔到MQ
                    if( !empty($result) && !empty($result['data']) ) 
                    {
                        $data = $result['data'];
                        foreach ($data as $item)
                        {
                            $_start_time = microtime(true);
                            $contentsArr = json_decode($item['contents'],true);
                            $result_mq = $this->pushSelfBbcMessage($user['channel_id'], $contentsArr['MessageType'], $contentsArr['data']);
                            echo json_encode($result_mq,JSON_UNESCAPED_UNICODE).PHP_EOL;
                            echo "耗时:". bcsub((string)microtime(true), (string)$_start_time,3).PHP_EOL;
                        }
                    }
                    
                    $wg->done();
                });
            }
            $wg->wait();
        }
    }
    
    /**
     * 查询BBC三方自研发渠道商用户
     */
    public function queryBbcOtherMessage()
    {
        while ( true )
        {
            $userList = Business::instance()
            ->select([
                'id',
                'channel_id',
                'appid',
                'open_url'
            ])
            ->where('is_channel',2)
            ->where('status',0)
            ->where('is_child_yc',0)
            ->where('has_template',1)
            ->where('template_price_tpl','!=','')
            ->where(function($query)
            {
                $query->whereRaw('user_tag is null or  user_tag = ""');
            })
            ->get()
            ->toArray();
    
            if( empty($userList) ) {
                sleep(3);
                continue;
            }
            
            $wg = new WaitGroup();
            $wg->add(count($userList));
            $wg_result = [];
            foreach ($userList as $user)
            {
                co(function() use ($wg,$user,& $wg_result)
                {
                    //开始拉取供应商接口商品增量同步消息
                    $response = Http::postJsonCurl(self::default_self_bbc_api,['appid'=>$user['appid']]);
                    $result   = !empty($response) ? json_decode($response,true) : [];
                    //取出数据之后，扔到MQ
                    if( !empty($result) && !empty($result['data']) )
                    {
                        $data = $result['data'];
                        foreach ($data as $item)
                        {
                            $notify_message = json_decode($item['contents'],true);
                            //推送标准下游三方接口
                            $postHead   = [
                                "User-Agent: server",
                                "Content-Type: application/json",
                            ];
                            
                            $_start_time = microtime(true);
                            $push_url = $user['open_url'];
                            $response = Http::apiRequest($push_url, $notify_message, 'post',$postHead, 30, false,'json');
                            $result   = !empty($response) ? json_decode($response,true) : [];
                            if( !empty($result) && isset($result['code']) && $result['code'] != Normal::SUC ) 
                            {
                                echo "推送数据异常【推送三方消息地址:{$push_url}，响应结果:{$response}】".PHP_EOL;               
                            } 
                            else
                            {
                                echo "推送数据结果【推送三方消息地址:{$push_url}，响应结果:{$response}】".PHP_EOL;
                            }
                            
                            echo "耗时:". bcsub((string)microtime(true), (string)$_start_time,3).PHP_EOL;
                        }
                    }
                    $wg->done();
                });
            }
            $wg->wait();
        }
    }
    
    /**
     * 开始推送
     * 仅仅支持接口api
     */
    public function push()
    {
        while ( true )
        {
            //获取系统并发推送限制
            $config = SystemCacheConfigService::instance()->getCacheInfoBySystemCode('channel_task_user_limit');
            $limit  = $config['TaskPushUserMessageCount'] ?? $this->pageSize ;
            $limit  = (int)$limit;
            $list = $this->queryMessageChangeLogLists($limit);
            if( empty($list) ) {
                sleep(self::SLEEP_TIMES_LIMIT);
            }
            
            //获取用户信息组
            $bids = array_column($list,'bid');
            $bids = array_unique($bids);
            $userList = Business::instance()->getUserInfoByBids($bids);
            $userList = array_column($userList,null,'bid');
            
            //大云仓需要排除 自营渠道和下级云仓
            $sysconfig = SystemCacheConfigService::instance()->getCacheInfoBySystemCode('sysconfig');
            $AppType   = $sysconfig['AppType'] ?? '';
            if( in_array($AppType,['shophglyc']) ) {
                foreach ($list as $index => $data_item) 
                {
                    $binfo = $userList[$data_item['bid']];
                    if( ($binfo['is_channel'] == 1 && $binfo['channel_id'] > 0) || $binfo['is_child_yc'] == 1 ) 
                    {
                        unset($list[$index]);
                    }
                }
            }
            
            if( empty($list) ) 
            {
                sleep(1);
                continue;
            }
            
            $wg = new WaitGroup();
            $wg->add(count($list));
            $wg_result = [];
            foreach ($list as $value)
            {
                co(function() use ($wg,$value,$userList,& $wg_result)
                {
                    usleep(5000);
                    $bid            = $value['bid'];
                    $binfo          = $userList[$bid];
                    $contents       = $value['contents'];
                    $message_id     = $value['id'];
                    $message_type   = $value['message_type'];
                    
                    if( $binfo['is_channel'] == 1 ) {
                        //$push_url = $this->getSelfBbcChannelURL();
                        $push_url = $value['push_url'];
                    } else if( $binfo['is_push_goods']) {
                        $push_url = $this->getThirdpartyChannelURL();
                    } else {
                        $push_url = $value['push_url'];
                    }
                    
                    //开始处理推送
                    $start          = microtime(true);
                    $result         = $this->pushSingleUserMessage($binfo,$contents,(string)$push_url,(string)$message_type,(string)$message_id);
                    $used_time      = bcsub((string)microtime(true), (string)$start,3);
                    
                    $date = getDateTime();
                    if( $used_time >= 1.5 ) {
                        echo "慢推送，时间：{$date}，推送商品类型 ： {$message_type}，bid : {$bid} ， 推送地址: {$push_url}，耗时:{$used_time} 秒 ".PHP_EOL;
                    }
                    
                    //记录推送结果
                    $wg_result[]    = [
                        'id'            => $message_id,
                        'bid'           => $bid,
                        'used_time'     => $used_time,
                        'message_type'  => $message_type,
                        'contents'      => $contents,
                        'push_status'   => isset($result['code']) && $result['code'] == Normal::SUC ? 2 : 1 ,
                        'push_result'   => json_encode($result,JSON_UNESCAPED_UNICODE),
                        'push_url'      => $push_url,
                        'created_at'    => date("Y-m-d H:i:s")
                    ];
                    
                    $wg->done();
                });
            }
            
            $wg->wait();
            
            //将写入message_user_skuchange_log日志异步写入MQ,由于数据包体大小问题
            foreach ($wg_result as $push_response)
            {
                $contents = $push_response['contents'];
                $message_type = $push_response['message_type'];
                unset($push_response['contents']);
                unset($push_response['message_type']);
                $this->pushSkuChangeLogMessage($push_response['bid'],(string)$push_response['id'],$message_type,$contents,$push_response);
            }
            
            //推送完成所有数据之后，批量删除当前商品消息
            $batch_ids = array_column($wg_result,'id');
            MessageSkuChangeLog::instance()->whereIn('id',$batch_ids)->delete();
        }
    }
    
    /**
     * 开始进行协程http并发推送
     * @param array $binfo
     * @param string $contents
     * @param string $push_url
     * @param string $message_type
     * @param string $message_id
     */
    public function pushSingleUserMessage(array $binfo, string $contents,string $push_url = null, string $message_type,string $message_id)
    {
        //解析当前推送地址
        if( empty($push_url) ) {
            return Normal::returnCode(Normal::FAIL,"地址格式不正确");
        }
        
        //解析当前推送地址
        $push_urls = parse_url($push_url);
        //如果不是标准推送地址，直接抛错
        if( empty($push_urls['scheme']) ) {
            return Normal::returnCode(Normal::FAIL,"地址格式不正确");
        }
        
        $pushMessage = @json_decode($contents,true);
        $MessageType = $pushMessage['MessageType'];
        $Message     = $pushMessage['data'];
        $messageId   = $pushMessage['messageId'] ?? '';
        //如果第三方下游属于捞取订单商户推送商品,走对应类的逻辑
        if( $binfo['is_push_goods'] == 1 )
        {
            if( !in_array($MessageType,['goodsChange','goodsSkuChange']) ) {
                return Normal::returnCode(Normal::SUC,"当前用户对应的商品消息类型无需推送");
            }
            $notify_message = [
                'binfo'         => $binfo,
                'MessageType'   => $MessageType,
                'data'          => $Message
            ];
        } else {
            $notify_message = $pushMessage;
        }
        
        //推送标准下游三方接口
        $postHead   = [
            "User-Agent: server",
            "Content-Type: application/json",
        ];
        $res            = Http::apiRequest($push_url, $notify_message, 'post',$postHead, 30, false,'json');
        $httpcode       = Http::getHttpCode();
        $push_result    = @json_decode((string)$res,true);
        if( isset($push_result['code']) ) {
            return $push_result;
        }
        return Normal::returnCode(Normal::FAIL,"返回JSON格式不正确【{$res}】");
    }
    
    /**
     * 获取自营BBC渠道地址
     * @return string
     */
    public function getSelfBbcChannelURL() 
    {
        $server_ip = env('SERVICE_IP');
        $server_rpc_yuncangmq_port = env('SERVER_RPC_YUNCANGMQ_PORT');
        $server_rpc_yuncangmq_port = $server_rpc_yuncangmq_port ? $server_rpc_yuncangmq_port : '9505';
        return "http://{$server_ip}:{$server_rpc_yuncangmq_port}/PushChannelMesage/RevSelfBbcChannel";
    }
    
    /**
     * 获取三方渠道地址
     * @return string
     */
    public function getThirdpartyChannelURL()
    {
        $server_ip = env('SERVICE_IP');
        $server_rpc_yuncangmq_port = env('SERVER_RPC_YUNCANGMQ_PORT');
        $server_rpc_yuncangmq_port = $server_rpc_yuncangmq_port ? $server_rpc_yuncangmq_port : '9505';
        return "http://{$server_ip}:{$server_rpc_yuncangmq_port}/PushChannelMesage/RevThirdpartyChannel";
    }
    
    /**
     * 获取云仓自营bbc渠道地址
     * @return string
     */
    public function getSelfYunCangBbcChannelURL()
    {
        $server_ip = env('SERVICE_IP');
        $server_rpc_yuncangmq_port = env('SERVER_RPC_YUNCANGMQ_PORT');
        $server_rpc_yuncangmq_port = $server_rpc_yuncangmq_port ? $server_rpc_yuncangmq_port : '9505';
        return "http://{$server_ip}:{$server_rpc_yuncangmq_port}/PushChannelMesage/selfYunCangBbcChannel";
    }
    
    /**
     * 商品新增或变更
     * 获取sid信息
     * @param int $cid
     * @param string $messageType
     * @param string $messageId
     * @param array $data
     * @param int $is_finish
     */
    public function pushSelfBbcMessage(int $cid,string $messageType,array $data,string $messageId = '',int $is_finish = 0)
    {
        try
        {
            $sid = env('bbc_sid');
            $sid = $sid ? $sid : 1;
            $message = [
                'sid'           => $sid,
                'cid'           => $cid,
                'data'          => $data,
                'MessageType'   => $messageType,
                'logid'         => $messageId,
                'IsFinished'    => $is_finish
            ];
            $message = new PushBbcMessageProducer($message);
            $producer = container()->get(Producer::class);
            $result = $producer->produce($message,true);
            if($result) {
                return Normal::returnCode(Normal::SUC,"发布成功");
            }
            return Normal::returnCode(Normal::FAIL,"发布消息队列失败");
        }
        catch ( \Throwable $th )
        {
            return Normal::returnCode(Normal::FAIL,"发布消息队列失败【{$th->getMessage()}】");
        }
    }
    
    /**
     * 推送三方消费数据队列
     * @param array $binfo
     * @param string $messageType
     * @param array $data
     * @return mixed
     */
    public function pushThirdBBCMessage(array $binfo,string $messageType,array $data)
    {
        try
        {
            $message = [
                'binfo'         => $binfo,
                'messageType'   => $messageType,
                'message'       => $data
            ];
            $message = new PushThirdPartyBbcMessageProducer($message);
            $producer = container()->get(Producer::class);
            $result = $producer->produce($message,true);
            if($result) {
                return Normal::returnCode(Normal::SUC,"发布成功");
            }
            return Normal::returnCode(Normal::FAIL,"发布消息队列失败");
        }
        catch ( \Throwable $th )
        {
            return Normal::returnCode(Normal::FAIL,"发布消息队列失败【{$th->getMessage()}】");
        }
    }
    
    /**
     * 推送下游skuLog日志数据
     * @param int $bid
     * @param string $message_id
     * @param string $message_type
     * @param string $message
     * @param array $push_response
     * @return mixed
     */
    public function pushSkuChangeLogMessage(int $bid,string $message_id,string $message_type,string $message, array $push_response)
    {
        try
        {
            $message = [
                'bid'           => $bid,
                'message_id'    => $message_id,
                'message_type'  => $message_type,
                'contents'      => $message,
                'push_response' => $push_response
            ];
            $message = new SkuUserChangeLogMessageProducer($message);
            $producer = container()->get(Producer::class);
            $result = $producer->produce($message,true);
            if($result) {
                return Normal::returnCode(Normal::SUC,"发布成功");
            }
            return Normal::returnCode(Normal::FAIL,"发布消息队列失败");
        }
        catch ( \Throwable $th )
        {
            return Normal::returnCode(Normal::FAIL,"发布消息队列失败【{$th->getMessage()}】");
        }
    }
}